package com.ruozedata.bigdata.hadoop

import java.net.URI

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{BlockLocation, FileSystem, LocatedFileStatus, Path, RemoteIterator}

/**
使用Scala调用HDFS API
 */
object HDFSApp {

    def main(args: Array[String]): Unit = {
        mkdir("/hdfsapi01")
        copyFileToHDFS("data/access.log", "/hdfsapi01")
        copyFileFromHDFS("/hdfsapi01/access.log", "data/data-io.txt")
        listFiles("/hdfsapi01")
    }


    // 获取FileSystem对象
    def getFileSystem(): FileSystem = {
        val configuration = new Configuration()
        configuration.set("dfs.client.use.datanode.hostname", "true")
        configuration.set("dfs.replication", "1");
        val uri = new URI("hdfs://hadoop000:8020")
        FileSystem.get(uri, configuration, "hadoop")
    }

    // 创建HDFS文件夹
    def mkdir(path: String): Unit = {
        val fileSystem: FileSystem = getFileSystem()
        val hdfsPath = new Path(path)
        if (fileSystem.exists(hdfsPath)) fileSystem.delete(hdfsPath, true)
        fileSystem.mkdirs(hdfsPath)
        fileSystem.close()
    }


    // 从本地复制文件至HDFS
    def copyFileToHDFS(filePath: String, hdfsPath: String): Unit = {
        val fileSystem: FileSystem = getFileSystem()
        val src = new Path(filePath)
        val dst = new Path(hdfsPath)
        fileSystem.copyFromLocalFile(src, dst)
        fileSystem.close()
    }


    // 从HDFS复制文件至本地
    def copyFileFromHDFS(hdfsFilePath: String, localPath: String): Unit = {
        val fileSystem: FileSystem = getFileSystem()
        val src = new Path(hdfsFilePath)
        val dst = new Path(localPath)
        fileSystem.copyToLocalFile(src, dst)
        fileSystem.close()
    }


    // 获取HDFS某个目录下所有文件
    def listFiles(hdfsPath: String): Unit = {
        val fileSystem: FileSystem = getFileSystem()

        val path = new Path(hdfsPath)
        val files: RemoteIterator[LocatedFileStatus] = fileSystem.listFiles(path, true)
        while (files.hasNext) {
            val fileStatus: LocatedFileStatus = files.next()
            val group: String = fileStatus.getGroup
            val owner: String = fileStatus.getOwner
            val blockSize: Long = fileStatus.getBlockSize
            val replication: Short = fileStatus.getReplication
            val permission: String = fileStatus.getPermission.toString
            val filePath: Path = fileStatus.getPath
            println(s"filePath = ${filePath}, permission = ${permission}, group = ${group}, owner = ${owner}, blockSize = ${blockSize}, replication = ${replication}")

        val blockLocations: Array[BlockLocation] = fileStatus.getBlockLocations
        println(s"block count = ${blockLocations.length}")
        }
        fileSystem.close()
    }
}
